/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.prediction.recalc;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Vector;
import org.apache.hadoop.io.WritableComparable;

/**
 *
 * @author nastya
 */
public class UserGrouppingAggregate implements WritableComparable<UserGrouppingAggregate> {

    private HashMap<GroupDistance, Vector<UserDistance>> map = new HashMap<GroupDistance, Vector<UserDistance>>();

    HashMap<GroupDistance, Vector<UserDistance>> getMap() {
        return map;
    }

    
    
    public UserGrouppingAggregate() {
    }

    public void add(String groupNum, double distance, Vector<UserDistance> users) {
        GroupDistance g = new GroupDistance(groupNum, distance);
        if (map.containsKey(g)) {
            Vector<UserDistance> got = map.get(g);
            got.addAll(users);
        } else {
            map.put(g, users);
        }
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(map.size());
        for (Map.Entry<GroupDistance, Vector<UserDistance>> entry : map.entrySet()) {
            writeEntry(out, entry);
        }
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        int mapSize = in.readInt();

        for (int i = 0; i < mapSize; i++) {
            readEntry(in);
        }
    }

    @Override
    public int compareTo(UserGrouppingAggregate t) {
        if (t == null) {
            return 1;
        }
        return ((Integer) map.size()).compareTo(t.map.size());
        
    }

    private void writeEntry(DataOutput out, Entry<GroupDistance, Vector<UserDistance>> entry) throws IOException {
        out.writeUTF(entry.getKey().getGroupId());
        out.writeDouble(entry.getKey().getDistance());
        out.writeInt(entry.getValue().size());
        for (UserDistance ud : entry.getValue()) {
            out.writeUTF(ud.getUserId());
            out.writeDouble(ud.getDistance());
        }
    }

    private void readEntry(DataInput in) throws IOException {
        String groupId = in.readUTF();
        double distance = in.readDouble();

        int vectorSize = in.readInt();
        Vector<UserDistance> v = new Vector<UserDistance>();

        for (int i = 0; i < vectorSize; i++) {
            String userId = in.readUTF();
            Double d = in.readDouble();
            v.add(new UserDistance(userId, distance));
        }

        add(groupId, distance, v);
    }
}
